package com.hudson.namesrv.handler;

import com.hudson.codec.domain.RemotingCommand;
import com.hudson.common.enums.MessageTypeEnum;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.StandardCharsets;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;

public class IdleMessageHandler extends SimpleChannelInboundHandler<RemotingCommand> {
    private static final Logger logger = LoggerFactory.getLogger(IdleMessageHandler.class);
    ConcurrentHashMap<String, CompletableFuture<Boolean>> promise = new ConcurrentHashMap<>();

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, RemotingCommand message) throws Exception {
        if (support(message.getType())) {
            promise.remove(channelHandlerContext.channel().id().asLongText());
        }
        channelHandlerContext.fireChannelRead(message);
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext channelHandlerContext, Object evt) {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state() == IdleState.READER_IDLE) {
                // 发送心跳包
                logger.debug("broker 查看连接是否关闭");
                channelHandlerContext.writeAndFlush(
                        RemotingCommand.builder()
                                .type(MessageTypeEnum.IDLE_MESSAGE.getType())
                                .body("你還好嗎".getBytes(StandardCharsets.UTF_8))
                                .build()).addListener(f -> {
                    if (f.isSuccess()) {
                        promise.put(channelHandlerContext.channel().id().asLongText(), new CompletableFuture<>());
                    } else {
                        BrokerMessageHandler.BROKER_CHANNEL.remove(channelHandlerContext.channel().id().asLongText());
                        channelHandlerContext.channel().close();
                    }
                }); // 或者发送自定义心跳消息
//                channelHandlerContext.writeAndFlush(new Message<>(MessageTypeEnum.IDLE_MESSAGE.getType(), "你还好吗?")).addListener(f -> {
//                    if (f.isSuccess()) {
//                        promise.put(channelHandlerContext.channel().id().asLongText(), new CompletableFuture<>());
//                    } else {
//                        BrokerMessageHandler.BROKER_CHANNEL.remove(channelHandlerContext.channel().id().asLongText());
//                        channelHandlerContext.channel().close();
//                    }
//                }); // 或者发送自定义心跳消息
            }
        }
    }

    public boolean support(String type) {
        return MessageTypeEnum.IDLE_MESSAGE.equals(MessageTypeEnum.fromType(type));
    }
}
